#include "reactor/event_loop.hpp"
#include "reactor/channel.hpp"
#include "reactor/epoll.hpp"
#include <sys/eventfd.h>
#include <unistd.h>
#include <cassert>
#include <iostream>

namespace reactor {

EventLoop::EventLoop()
    : _looping(false),
      _quit(false),
      _callingPendingFunctors(false),
      _threadId(std::this_thread::get_id()),
      _poller(new Epoll()),
      _wakeupFd(::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC))
{
    if (_wakeupFd < 0) {
        std::cerr << "Failed in eventfd" << std::endl;
        abort();
    }
    _wakeupChannel.reset(new Channel(this, _wakeupFd));
    _wakeupChannel->setReadCallback(std::bind(&EventLoop::handleRead, this));
    _wakeupChannel->enableReading();
}

EventLoop::~EventLoop() {
    _wakeupChannel->disableAll();
    _wakeupChannel->remove();
    ::close(_wakeupFd);
}

void EventLoop::loop() {
    assertInLoopThread();
    _looping = true;
    _quit = false;

    while (!_quit) {
        std::vector<Channel*> activeChannels;
        _poller->poll(activeChannels);
        for (Channel* channel : activeChannels) {
            channel->handleEvent();
        }
        doPendingFunctors();
    }
    _looping = false;
}

void EventLoop::quit() {
    _quit = true;
    if (!isInLoopThread()) {
        wakeup();
    }
}

void EventLoop::runInLoop(Functor cb) {
    if (isInLoopThread()) {
        cb();
    } else {
        queueInLoop(std::move(cb));
    }
}

void EventLoop::queueInLoop(Functor cb) {
    {
        std::lock_guard<std::mutex> lock(_mutex);
        _pendingFunctors.push_back(std::move(cb));
    }
    if (!isInLoopThread() || _callingPendingFunctors) {
        wakeup();
    }
}

void EventLoop::updateChannel(Channel* channel) {
    assertInLoopThread();
    _poller->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel) {
    assertInLoopThread();
    _poller->removeChannel(channel);
}

void EventLoop::assertInLoopThread() {
    if (!isInLoopThread()) {
        std::cerr << "EventLoop::assertInLoopThread() - not in loop thread" << std::endl;
        abort();
    }
}

void EventLoop::wakeup() {
    uint64_t one = 1;
    ssize_t n = ::write(_wakeupFd, &one, sizeof(one));
    if (n != sizeof(one)) {
        std::cerr << "EventLoop::wakeup() writes " << n << " bytes instead of 8" << std::endl;
    }
}

void EventLoop::handleRead() {
    uint64_t one = 1;
    ssize_t n = ::read(_wakeupFd, &one, sizeof(one));
    if (n != sizeof(one)) {
        std::cerr << "EventLoop::handleRead() reads " << n << " bytes instead of 8" << std::endl;
    }
}

void EventLoop::doPendingFunctors() {
    std::vector<Functor> functors;
    _callingPendingFunctors = true;
    {
        std::lock_guard<std::mutex> lock(_mutex);
        functors.swap(_pendingFunctors);
    }
    for (const Functor& functor : functors) {
        functor();
    }
    _callingPendingFunctors = false;
}

} // namespace reactor
